
#include "platform.h"
#include "process.h"

#include "job_queue.h"

using namespace threading;

// #define COM_INITIALIZATION

void job_notify_type::notify() // this gets called by 
{
	switch (notification) {
	case postmessage:
		::PostMessage( msg.hwnd, msg.message, msg.wParam, msg.lParam );
		break;
	case setevent:
		SetEvent( (HANDLE)msg.lParam );
		break;
	}
}

job_notify_type::job_notify_type()
{
	repost = false;
	notification = none;
	shouldDelete = false;
}

job_notify_type::job_notify_type(const job_notify_type& _src)
{
	repost = _src.repost;
	msg.hwnd = _src.msg.hwnd;
	msg.message = _src.msg.message;
	msg.wParam = _src.msg.wParam;
	msg.lParam = _src.msg.lParam;
	notification = _src.notification;
	shouldDelete = _src.shouldDelete;
}

void job_notify_type::setSignal( HANDLE signal )
{
	notification = setevent;
	msg.lParam = (DWORD)(signal);
}

void job_notify_type::setPostMsg( HWND _hwnd, UINT _message, WPARAM _wParam, LPARAM _lParam )
{
	notification = postmessage;
	msg.hwnd = _hwnd;
	msg.message = _message;
	msg.wParam = _wParam;
	msg.lParam = _lParam;
}

job_notify_type::~job_notify_type()
{
	;
}

void job_notify_type::operator = (const job_notify_type& _src)
{
	repost = _src.repost;
	msg.hwnd = _src.msg.hwnd;
	msg.message = _src.msg.message;
	msg.wParam = _src.msg.wParam;
	msg.lParam = _src.msg.lParam;
	notification = _src.notification;
	shouldDelete = _src.shouldDelete;
}


// ----------------------------------------------------------------------------

job_type::job_type( )
{
	next = NULL;
	::ZeroMemory( &ovp, sizeof(ovp) );
}

job_type::~job_type()
{
	;
}

job_notify_type job_type::execute( job_queue_type *_callingQueue, DWORD _bytesTransferred, BOOL _success ) 
{
	job_notify_type t;

	return t;
}

// -------------------------------------------------------------------------------

finishjob_type::finishjob_type( )
{
	handle = ::CreateEvent( NULL, FALSE, FALSE, NULL );
}

finishjob_type::~finishjob_type()
{
	::CloseHandle( handle );
}

job_notify_type finishjob_type::execute( job_queue_type *_callingQueue, DWORD _bytesTransferred, BOOL _success )
{
	job_notify_type jobNotify;

	jobNotify.setSignal( handle );

	return jobNotify;
}

// -------------------------------------------------------------------------------
// simple wrapper for interlocked increment

lockable_type::lockable_type( int _spin_count )
{
	::InitializeCriticalSectionAndSpinCount( &section, _spin_count );
	lock_count = 0;
}

// these two do not assign the critical section to work ok with the lock_library

lockable_type::lockable_type( const lockable_type& _src ) // evil if _src locked
{
	::InitializeCriticalSection( &section );
	lock_count = 0;
}

lockable_type lockable_type::operator = ( const lockable_type& _src ) // evil if _src locked
{
	::InitializeCriticalSection( &section );
	lock_count = 0;
	return *this;
}

lockable_type::lockable_type()
{
	::InitializeCriticalSection( &section );
	lock_count = 0;
}

void lockable_type::initialize()
{
	::InitializeCriticalSection( &section );
	lock_count = 0;
}

lockable_type::~lockable_type()
{
	::DeleteCriticalSection( &section );
}

void lockable_type::lock()
{
	::EnterCriticalSection( &section );
	lock_count++;
}

void lockable_type::unlock()
{
	lock_count--;
	::LeaveCriticalSection( &section );
}

bool lockable_type::locked()
{
	return lock_count > 0;
}

// ---

locker_type::locker_type( lockable_type *_lockable )
{
	lockable = _lockable;
	lockable->lock();
}

locker_type::~locker_type()
{
	if (lockable) {
		lockable->unlock();
	}
}

// ---

job_queue_type::job_queue_type()
{
	ioCompPort = NULL;
	num_outstanding_jobs = 0;
	empty_queue_event = ::CreateEvent( NULL, FALSE, TRUE, NULL );
}

job_queue_type::~job_queue_type()
{
	::CloseHandle( empty_queue_event );
}

void job_queue_type::start( int _numThreads )
{
	int i;

	shutDownOrdered = false;

	int threadCount = job_queue_type::getThreadCount();

	if (threadCount && _numThreads == 0) {
		_numThreads = threadCount;
	}

	if (_numThreads > maxWorkerThreads || _numThreads == 0) _numThreads = maxWorkerThreads;

	if (!ioCompPort) {

		numWorkerThreads = _numThreads;

		ioCompPort = CreateIoCompletionPort( INVALID_HANDLE_VALUE, NULL, 0, _numThreads );

		if (ioCompPort) {
			for (i = 0; i < numWorkerThreads; i++) {
				threads[ i ] = (HANDLE)_beginthreadex( NULL, NULL, jobQueueThread, (void *)(this), 0, &threadIds[ i ] );
			}
		}
	}
}

void job_queue_type::postJobMessage( job_type *_jobMessage )
{
	LONG result;
	::InterlockedIncrement( (LONG *)&num_outstanding_jobs );
	::ResetEvent( empty_queue_event );
	result = ::PostQueuedCompletionStatus( ioCompPort, 0, 0, (LPOVERLAPPED)( _jobMessage ) );
}

void job_queue_type::shutDown()
{
	int i;
	DWORD timeout;

	// keep sending shutdown messages until all of the queues terminate!

	shutDownOrdered = true;

	for (timeout = WAIT_TIMEOUT; timeout == WAIT_TIMEOUT; timeout = WaitForMultipleObjects( numWorkerThreads, threads, 200, FALSE ) ) 
	{
		postJobMessage( &shutDownJob );
	}

	::CloseHandle( ioCompPort );

	for (i = 0; i < numWorkerThreads; i++) {
		::CloseHandle( threads[ i ] );
		threads[ i ] = NULL;
		threadIds[ i ] = 0;
	}

	ioCompPort = NULL;
}

void job_queue_type::kill()
{
	int i;

	shutDownOrdered = true;

	if (!ioCompPort) {
		for (i = 0; i < numWorkerThreads; i++) {
			::TerminateThread( threads[ i ], 0 );
			::CloseHandle( threads[ i ] );
			threads[ i ] = NULL;
			threadIds[ i ] = 0;
		}
	}

	::CloseHandle( ioCompPort );
	ioCompPort = NULL;
}

unsigned int __stdcall job_queue_type::jobQueueThread( void *_arg )
{
	job_queue_type *jobQueue = (job_queue_type *)(_arg);

	LPOVERLAPPED lpov;
	job_type *job;

	BOOL success, shuttingDown;
	DWORD bytesTransferred;
	ULONG_PTR compKey;

	shuttingDown = FALSE;
	job_notify_type jobNotify;

#ifdef COM_INITIALIZATION
//	::CoInitializeEx( NULL, COINIT_MULTITHREADED );
	CoInitialize( NULL );
#endif

	while (!shuttingDown) {
		success = ::GetQueuedCompletionStatus( jobQueue->ioCompPort, &bytesTransferred, &compKey, &lpov, INFINITE );
		job = (job_type *)lpov;

		if (job == &(jobQueue->shutDownJob)) {
			shuttingDown = true;
		} 
		else if (job) {
			jobNotify = job->execute( jobQueue, bytesTransferred, success );
			jobNotify.notify();
			if (jobNotify.repost && (!jobQueue->wasShutDownOrdered())) {
				jobQueue->postJobMessage( job );
			}
			if (jobNotify.shouldDelete) {
				delete job;
			}
		}

		LONG numJobs = ::InterlockedDecrement( (LONG *)(&jobQueue->num_outstanding_jobs) );

		if (!numJobs) {
			::SetEvent( jobQueue->empty_queue_event );
		}

	}

#ifdef COM_INITIALIZATION
	::CoUninitialize();
#endif

	return 0;
};

void job_queue_type::waitForThreadFinished()
{
	finishjob_type finishJob;

	postJobMessage( &finishJob );
	::WaitForSingleObject( finishJob.handle, INFINITE );
}

void job_queue_type::waitForEmptyQueue()
{
	::WaitForSingleObject( empty_queue_event, INFINITE );
}

int job_queue_type::numberOfProcessors()
{
	return strtol( getenv( "NUMBER_OF_PROCESSORS" ), NULL, 10 );
}
